package site.luojie.netty.rpc.consumer;

import com.alibaba.fastjson.JSON;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import site.luojie.netty.rpc.common.Server;
import site.luojie.netty.rpc.common.UserService;

import java.util.*;
import java.util.stream.Collectors;


/**
 * 服务消费者启动主程序类
 *
 * @author jie.luo
 * @since 2020-07-06 23:54
 **/
public class RpcConsumerZk {

    public final static String root = "/servers";
    public final static String ACTION_PATH = "/action";

    public static void loadProvider(String zkConnectString) throws Exception {

        // 建立zk链接
        ZkClient zkClient = new ZkUtil().build(zkConnectString).getZkClient();
        // 建立响应时间记录父节点
        boolean exists = zkClient.exists(ACTION_PATH);
        if (!exists) {
            zkClient.createPersistent(ACTION_PATH);
        }

        // 获取在线服务端信息子节点列表
        List<String> children = zkClient.getChildren(root);

        refresh(zkClient, children);

        zkClient.subscribeChildChanges(root, new IZkChildListener() {
            @Override
            public void handleChildChange(String path, List<String> childes) throws Exception {

                refresh(zkClient, childes);

            }
        });

    }

    /**
     * 根据子节点列表建立与服务端的链接
     */
    public static void refresh(ZkClient zkClient, List<String> children) throws InterruptedException {

        List<Server> servers = new ArrayList<>();
        Server server = null;

        for (String child : children) {

            String serverJson = zkClient.readData(root + "/" + child);

            server = JSON.parseObject(serverJson, Server.class);
            server.setKey(child);

            servers.add(server);
        }

        onRefresh(zkClient, servers);
    }

    public static void onRefresh(ZkClient zkClient, List<Server> servers) throws InterruptedException {

        if (servers.size() > ServerPool.getServerNum()) {
            // 服务端上线，向池子中添加服务端连接
            for (Server server : servers) {
                if (!ServerPool.containsServer(server.getKey())) {
                    String actionServerPath = ACTION_PATH + "/" + server.getKey();
                    boolean exists = zkClient.exists(actionServerPath);
                    if (!exists) {
                        zkClient.createPersistent(actionServerPath);
                    }
                    UserService userService = (UserService) new RpcConsumer().build(server.getHost(), server.getPort()).proxy(UserService.class, server.getKey());
                    ServerPool.putServer(server.getKey(), userService);
                }
            }
        } else if (servers.size() < ServerPool.getServerNum()) {
            // 服务端下线，从池子中剔除无效连接
            // 获取有效连接的key
            Set<String> effectiveServers = servers.stream().map(server -> server.getKey()).collect(Collectors.toSet());
            // 从池子中删除无效连接
            Iterator<String> iterator = ServerPool.getServersKey().iterator();
            while (iterator.hasNext()) {
                String key = iterator.next();
                if (!effectiveServers.contains(key)) {
                    ServerPool.removeServer(key);
                }
            }
        } else {
            // 服务端数量与池子数量一致
            return;
        }

    }

    /**
     * 服务路由
     */
    public static UserService routeServer() throws Exception {
        return ServerPool.getServer(routeRule());
    }

    public static String routeRule() throws Exception {
        ZkClient zkClient = new ZkUtil().getZkClient();

        List<String> children = zkClient.getChildren(ACTION_PATH);

        if (null == children || children.size() == 0) {
            throw new Exception("还未与服务端建立链接");
        }

        String submit = null;
        for (String child : children) {
            String actionJson = zkClient.readData(ACTION_PATH + "/" + child);
            if (null == actionJson) {
                submit = child;
                break;
            }
        }

        if (null == submit) {

            List<ActionDto> actionDtos = new ArrayList<>();
            for (String child : children) {
                String actionJson = zkClient.readData(ACTION_PATH + "/" + child);
                ActionDto actionDto = JSON.parseObject(actionJson, ActionDto.class);

                // 验证记录时间是否过期，有效时间5秒
                long currentTimeMillis = System.currentTimeMillis();
                if (currentTimeMillis - actionDto.getCreateTime() > 5000) {
                    actionDto.setResponseTime(0);
                }

                actionDtos.add(actionDto);
            }

            List<ActionDto> collect = actionDtos.stream().sorted(Comparator.comparing(ActionDto::getResponseTime)).collect(Collectors.toList());

            submit = collect.get(0).getPath();
        }

        return submit;
    }

}
